package cn.myzf.fort.fortaccess.autoconfigure;

import cn.myzf.fort.access.accesscommon.autoconfigure.JschProperties;
import cn.myzf.fort.access.accesscommon.core.ChannelProperties;
import cn.myzf.fort.access.accesscommon.core.ExecutorServices;
import cn.myzf.fort.access.accesscommon.core.Utils;
import cn.myzf.fort.access.accesscommon.dto.HostInfo;
import cn.myzf.fort.access.accesscommon.dto.SubHost;
import cn.myzf.fort.access.accesscommon.utils.SpringUtil;
import cn.myzf.fort.access.exe.PLink;
import com.jcraft.jsch.JSch;
import com.jcraft.jsch.JSchException;
import com.jcraft.jsch.Session;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.net.telnet.TelnetClient;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.boot.autoconfigure.AutoConfigureOrder;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.Ordered;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.PrintStream;
import java.net.DatagramSocket;
import java.net.Socket;
import java.net.SocketException;
import java.util.*;
import java.util.concurrent.*;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Collectors;

@Configuration
@AutoConfigureOrder(Ordered.HIGHEST_PRECEDENCE)
@EnableConfigurationProperties(JschProperties.class)
@Slf4j
public class JschAutoConfiguration implements InitializingBean, DisposableBean {



	private final JschProperties jschProperties;

	private static final ConcurrentHashMap<String, Session> sshSessionMap = new ConcurrentHashMap<>();

	private static final ConcurrentHashMap<String, HostInfo> VIRTUAL_HOST = new ConcurrentHashMap<>();

	private static final ConcurrentHashMap<String, Process> sock5_Process = new ConcurrentHashMap<>();

	private static final  Set<Integer> localPortForwarders = new CopyOnWriteArraySet<>();

	private ExecutorService executorService = new ExecutorServices().createExecutorService();

	private ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();

	private String LOCALHOST = "127.0.0.1";

	private List<HostInfo> list;

	//private final static Map<String, Set<PortForwardDTO>> forwardMap = new HashMap<>();

	private Session session;

	public JschAutoConfiguration(JschProperties jschProperties) {
		initList();
		this.jschProperties = jschProperties;
	}


	public void start() throws Exception {

		JSch jsch = new JSch();
		/*由于主机信息一般不会变，所以这里存储在数据库，先从本地数据库存*/
		session = jsch.getSession(jschProperties.getProxyUser(), jschProperties.getProxyHost(),
				jschProperties.getProxyPort());
		session.setPassword(jschProperties.getProxyPassword());
		session.setConfig("StrictHostKeyChecking", jschProperties.getStrictHostKeyChecking());
		log.info(String.format("Jsch_AutoConfiguration connect to:::host: %s , port: %d, user: %s\n",
				jschProperties.getProxyHost(), jschProperties.getProxyPort(), jschProperties.getProxyUser()));
		session.setDaemonThread(true);//设置守护进程，系统退出随之session销毁
		session.connect();

		log.info(String.format("Jsch_AutoConfiguration::: %s \n", session.getServerVersion()));// 打印SSH服务器版本信息

		/** 添加conn映射
		 *  ip ==》session
		 * */
		addConnectMapping(jschProperties.getProxyHost(),session);
		allSourceList();
		if(listIsNotNull(list)){
			for (HostInfo h : list) {
				/*开启一个本地空闲端口*/
				int localPort = h.getVirtualPort()>0?h.getVirtualPort():new DatagramSocket(0).getLocalPort();
				h.setVirtualPort(localPort);
				/** 添加本地虚拟端口映射
				 * ip ==> hostinfo虚拟机信息
				 * 跳板机 ==》 hostinfo配置的端口转发
				 * */
				addHostForwordL(h);
				log.info(String.format("Jsch_AutoConfiguration:::localhost:%d ===>>> %s:%d 应用名：%s", localPort,
						h.getHost(), h.getPort(),h.getAgent()));
				if (h.isSock5()){
					executorService.submit(new PLinkImplRunnable(h));
				}
				if (h.getSubHost()!=null && h.getSubHost().size() > 0) {
					Session subSession = null;
					JSch subJsh = new JSch();
					subSession = subJsh.getSession(h.getUser(),LOCALHOST,h.getVirtualPort());
					subSession.setPassword(h.getPass());
					subSession.setConfig("StrictHostKeyChecking", jschProperties.getStrictHostKeyChecking());
					log.info(String.format("SubJsch_AutoConfiguration connect to:::host: %s , port: %d, user: %s\n",
							LOCALHOST, h.getVirtualPort(), h.getUser()));
					subSession.setDaemonThread(true);//设置守护进程，系统退出随之session销毁
					subSession.connect();
					for (SubHost subHost : h.getSubHost()) {
						subSession.setPortForwardingL(subHost.getSubVport(),subHost.getSubHost(),subHost.getSubPort());
					}
				}
			}

		}

		//监控跳板机ssh线程 （所有的操作都是基于它的，所以要监控下 掉线【拔网线、跳板机重启等情况】可以重启，默认尝试3次）
		startDetectPort();
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		start();
	}


	private void addHostForwordL(HostInfo h){
		addHostINFO(h.getHost(),h);
		forwordL(session,h.getVirtualPort(), h.getHost(), h.getPort());
	}

	private void allSourceList(){
		Map<String, ChannelProperties> sourceHostInfo = SpringUtil.getApplicationContext().getBeansOfType(ChannelProperties.class);
		sourceHostInfo.values().stream().forEach(source -> {
			List<HostInfo> soureList = source.list();
			if (listIsNotNull(soureList)) list.addAll(duplicate(soureList));
		});
		//解决treeset 比较重复bug 通过二叉树，只比较最大的，当两个集合合并的时候，可能会出现相同元素，不再参与比价问题
		//这里先把list通过名字分组，这样相同的元素连续，比较的时候就不会出现，同样的host地址，但是大小不同，从而少去重的问题
		sortByType();
		list = duplicate(list);
	}

	private void initList(){list = new CopyOnWriteArrayList<>(); }





	public  <T> Predicate<T> distinctByKey(Function<? super T, ?> keyExtractor) {
		ConcurrentHashMap<Object, Boolean> map = new ConcurrentHashMap<>(16);
		return t -> map.putIfAbsent(keyExtractor.apply(t), Boolean.TRUE) == null;
	}

	//去重
	private List<HostInfo> duplicate(List<HostInfo> list){

		Set<HostInfo> hostInfos = new TreeSet<>(new Comparator<HostInfo>() {
			@Override
			public int compare(HostInfo o1, HostInfo o2) {
				//当主机相同的时候 去重，当某一条有数据的时候就有有数据的
				int result  = 1;
				//当ip 虚拟端口
				if (StringUtils.equals(o1.getHost(),o2.getHost())) {
					if (o1.getVirtualPort()>0 || o2.getVirtualPort() >0 ) {
						o2.setVirtualPort(o1.getVirtualPort()>o2.getVirtualPort()?o1.getVirtualPort():o2.getVirtualPort());
					}
					if (o1.isSock5() || o2.isSock5()) {
						o2.setSock5(true);
						o2.setSock5Port(o1.isSock5()?o1.getSock5Port():o2.getSock5Port());//o1开启取o1的端口，其他都取o2的端口 这个其实也无所谓的只要有端口即可
					}
					if (o1.getSubHost()!=null || o2.getSubHost() != null) {
						o2.setSubHost(o1.getSubHost()!=null?o1.getSubHost():o2.getSubHost());
					}
					result = 0;//两者相等默认取o2
				}
				return result;
			}


		});
		hostInfos.addAll(list);
		return new ArrayList<HostInfo>(hostInfos);
	}

	private boolean listIsNotNull(List list){
		if (list!=null && list.size()>0) return true;
		return false;
	}


	//根据相同属性对list 分组， 便于查看日志
	public void sortByType() {
		list = list.stream().sorted(Comparator.comparing(HostInfo::getAgent)).collect(Collectors.toList());
	}

	/**
	 * 主机和ssh连接的映射
	 *
	 */
	public void addConnectMapping(final String ip, final Session session) {
		sshSessionMap.put(ip, session);
	}





	public  boolean checkVirtualStatus(int virtualPort) {
		boolean f = false;
		try {
			TelnetClient telnetClient = new TelnetClient("VT200");  //指明Telnet终端类型，否则会返回来的数据中文会乱码
			telnetClient.setDefaultTimeout(5000); //socket延迟时间：5000ms
			telnetClient.connect(LOCALHOST,virtualPort);  //建立一个连接,默认端口是23
			f = telnetClient.isConnected();
			PrintStream pStream = new PrintStream(telnetClient.getOutputStream());  //写命令的流
			pStream.println("exit"); //写命令
			pStream.flush(); //将命令发送到telnet Server
			if(null != pStream) {
				pStream.close();
			}
			telnetClient.disconnect();
		} catch (SocketException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return f;
	}

	/**
	 * 主机和ssh连接的映射
	 *
	 */
	public void addHostINFO(final String ip, final HostInfo hostInfo) { VIRTUAL_HOST.put(ip, hostInfo); }


	public void forwordL(Session session,int port,String connect_host,int connct_port){
		try {
			int lport = session.setPortForwardingL(port, connect_host, connct_port);
			localPortForwarders.add(lport);
		} catch (JSchException e) {
			if (e.getMessage().contains(String.format("local port 127.0.0.1:%s cannot be bound.",port)))
			{
				String pid = findPid(port);
				if (StringUtils.isNotBlank(pid))
					Utils.killProcess(pid);
				forwordL(session,port,connect_host,connct_port);
			}

			e.printStackTrace();
		}
	}


	public String findPid(int port){
		try {
		return 	Utils.checkListening(String.valueOf(port));
		}catch (IOException e) {
			e.printStackTrace();
		}
		return "";
	}

	public void forwordR(Session session,int port,String connect_host,int connct_port){
		try {
			session.setPortForwardingR(10933,"10.6.110.133",10933);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * 本地转发请求的端口如果长时间不用，会在一定时间内退出监听， 
	 * 从而使转发请求失败， 所以启动一个线程定期(每隔30s)给这个端口发请求。
	 */

	private int retryTime;
	private void startDetectPort() throws Exception {
		// 创建一个流套接字并将其连接到本地转发端口上
			// 向服务器端发送数据
			Runnable runnable = new Runnable() {
				@SneakyThrows
				public void run() {
					try {
						Socket socket = new Socket(jschProperties.getProxyHost(), jschProperties.getProxyPort());
							 DataOutputStream out = new DataOutputStream(socket.getOutputStream());
						log.info(String.format("%1$tY-%1$tm-%1$td %tT ping---->%s:%d\n", new Date(), jschProperties.getProxyHost(), 22));
						out.writeUTF("ping");
					} catch (Exception e) {
						log.error(String.format("%1$tY-%1$tm-%1$td %tT ping---->%s:%d request detection exception, exit ping. errmsg: %s\n", new Date(),
								jschProperties.getProxyHost(), jschProperties.getProxyPort(), e.getMessage()));
						if (retryTime<3) {
							restart();
							retryTime++;
							return;
						}
							log.error("The maximum number of retries has been reached, the springboard machine is still not available!!!");
							System.exit(1);

					}
				}
			};
			/*检查虚拟端口是否工作正常，网络抖动引发掉线等*/
		Runnable runnable1 = new Runnable() {
			@SneakyThrows
			public void run() {
				try {
					localPortForwarders.stream().forEach(virtual_port -> {
						boolean b = checkVirtualStatus(virtual_port);
						if (!b) {
							log.info(String.format("本地虚拟端口%s工作异常",virtual_port));
							//执行端口重启
							try {
								restart();
							} catch (Exception e) {
								e.printStackTrace();
							}
						}
					});


				} catch (Exception e) {

				}
			}
		};

		service.scheduleAtFixedRate(runnable, 60, 60, TimeUnit.SECONDS);
		service.scheduleAtFixedRate(runnable1, 30, 5, TimeUnit.SECONDS);

	}

	public synchronized void restart() throws Exception {
		if (!localPortForwarders.isEmpty() && !sshSessionMap.isEmpty())
		shutdown();
		session = null;
		list = new CopyOnWriteArrayList<>();
		start();
	}

	@Override
	public void destroy() throws Exception {
		service.shutdownNow();
		shutdown();
		System.out.println("Jsch_AutoConfiguration::: destory connection");
	}

	public static void main(String[] args) {
		PriorityQueue<Object> objects = new PriorityQueue<>();
		objects.add(1);
		objects.add(2);
		objects.add(3);
		while (!objects.isEmpty())
			System.out.println(objects.poll());
		int size = 3;
		size  -=1;
		int i = size >>> 1;
		System.out.println(size);
	}


	private void shutdown() throws Exception {
		/*关闭端口转发*/
		localPortForwarders.forEach(lport ->accept(lport));
		localPortForwarders.clear();

		/*倒序关闭conn 由于链接存在先后顺序*/
		reverse(sshSessionMap).values().forEach(session -> session.disconnect());
		if (!sock5_Process.isEmpty())
	//	sock5_Process.values().forEach(process -> process.destroy());
		log.info("Genymed_AutoConfiguration::: destory connection");
	}

	private void accept(Integer lport) {
		try {
			if (session!=null) { session.delPortForwardingL(lport); }
		} catch (JSchException e) {
			e.printStackTrace();
		}
	}

	/*逆序map*/
	public  <K, V> Map<K, V> reverse(Map<K,V> map){
		ListIterator<Map.Entry<K, V>> li = new ArrayList<Map.Entry<K, V>>(map.entrySet()).listIterator(map.size());
		ConcurrentHashMap<K, V> reverseMap = new ConcurrentHashMap<>();
		while(li.hasPrevious()) {
			Map.Entry<K, V> entry = li.previous();
			reverseMap.putIfAbsent(entry.getKey(),entry.getValue());
		}
		return reverseMap;
	}


	/**
	 * 由于plink会阻塞线程，监听端口，子线程来执行代理任务
	 * sock5线程任务
	 * */
	 class PLinkImplRunnable implements Runnable {
		private HostInfo h;
		public PLinkImplRunnable(HostInfo h) {
			this.h = h ;
		}
		@Override
		public void run() {
			PLink pLink = new PLink()
					.sshPort(String.valueOf(h.getVirtualPort()))
					.sshUser(h.getUser())
					.sshHost(LOCALHOST)
					.sock5Host(LOCALHOST)
					.sock5Port(h.getSock5Port())
					.sshPw(h.getPass())
					.logName(h.getAgent())
					;
			try {
				pLink.execute();
			//	if (h.isSock5())
				//sock5_Process.put(h.getSock5Port(),execute.getProcess());
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}

}
